package com.hmeter.hadoop.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * Component name: HBaseDeleteSampler
 * Class used for HBase Delete operations that supports following functionalities:
 * 1. Delete column qualifiers in HBase row (delete list of column qualifiers in particular row on latest timestamp)
 * 2. Delete column families data in HBase row (delete list of column families content in particular row on latest timestamp)
 * 
 * @author Bakir Jusufbegovic
 * @since 0.1
 * @version 0.1
 */
public class HBaseDeleteSamplerUtility {
	
private HBaseUtility hbaseUtility = null;
	
	public HBaseDeleteSamplerUtility(String hbaseZooKeeperQuorum, String hbaseTable) throws IOException 
	{
		hbaseUtility = new HBaseUtility(hbaseZooKeeperQuorum, hbaseTable);
	}
	
	public HBaseUtility getHBaseUtility()
	{
		return hbaseUtility;
	}
	
	public String deleteColumnQualifiersFromHBase(String rowKey, String fullColumnQualifiers, String filterColumnFamilies, String delimiter) throws IOException
	{	
		String response = "";
		
		List<Delete> deletes = new ArrayList<Delete>();
		Delete delete = new Delete(Bytes.toBytes(rowKey));
		
		if (delimiter.equals(""))
		{
			delimiter = delimiter.concat(",");
			fullColumnQualifiers = fullColumnQualifiers.concat(",");
			filterColumnFamilies = filterColumnFamilies.concat(",");
		}

		ArrayList<String> parsedList = hbaseUtility.parseElementsToList(fullColumnQualifiers, delimiter);
		long timestamp = hbaseUtility.getMostRecentTimestamp(rowKey, hbaseUtility.parseElementsToList(filterColumnFamilies, delimiter));		
		String[] columnFamilyColumnValueToken = null;

		try
		{			
			for (int i=0; i<parsedList.size(); i++)
			{
				if(parsedList.get(i).contains(":")) {					
					columnFamilyColumnValueToken = parsedList.get(i).split(":");
					delete.deleteColumn(Bytes.toBytes(columnFamilyColumnValueToken[0]), Bytes.toBytes(columnFamilyColumnValueToken[1]), Long.valueOf(timestamp));
				}
				deletes.add(delete);
			}		
			hbaseUtility.getHTable().delete(deletes);
			response = "Deleted list of column qualifiers: " + fullColumnQualifiers;
		}
		catch (IOException ex) {
			response = ex.getMessage();
		}
		
		return response;
	}
	
	public String deleteColumnFamiliesDataFromHBase(String rowKey, String columnFamilies, String filterColumnFamilies, String delimiter) throws IOException
	{	
		String response = "";
		
		List<Delete> deletes = new ArrayList<Delete>();
		Delete delete = new Delete(Bytes.toBytes(rowKey));

		ArrayList<String> parsedList = hbaseUtility.parseElementsToList(columnFamilies, delimiter);
		long timestamp = hbaseUtility.getMostRecentTimestamp(rowKey, hbaseUtility.parseElementsToList(filterColumnFamilies, delimiter));		

		try
		{			
			for (int i=0; i<parsedList.size(); i++)
			{
				delete.deleteFamily(Bytes.toBytes(parsedList.get(i)), Long.valueOf(timestamp));
				deletes.add(delete);
			}		
			hbaseUtility.getHTable().delete(deletes);
			response = "Deleted list of column families: " + columnFamilies;
		}
		catch (IOException ex) {
			response = ex.getMessage();
		}
		
		return response;
	}
}
